package com.xdja.kafka.hdfs.sink;
import java.io.IOException;
import java.util.*;
import com.alibaba.fastjson.JSON;
import com.xdja.kafka.constants.GlobalConstant;
import com.xdja.kafka.hdfs.sink.bean.ConnectHdfsConfig;
import com.xdja.kafka.hdfs.sink.bean.HuaweiLoginBean;
import com.xdja.kafka.hdfs.sink.task.TextSinkTask;
import com.xdja.kafka.hdfs.sink.task.ParquetSinkTask;
import com.xdja.kafka.util.LoginUtil;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
/**
 * 向hdfs写入的connector
 */
public class HdfsSinkConnector extends SinkConnector {
    private static final Logger log = LoggerFactory.getLogger(HdfsSinkConnector.class);
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
            .define(GlobalConstant.HDFS_URL, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "hdfs url")
            .define(GlobalConstant.HDFS_PATH, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "hdfs path");
    /**
     * connector hdfs的配置信息
     */
    public static ConnectHdfsConfig connectHdfsConfig;

    /**
     * 设定版本
     * @return
     */
    @Override
    public String version() {
        String version="1.0.0";
        return version;
    }

    /**
     * 启动
     * @param props
     */
    @Override
    public void start(Map<String, String> props) {
        log.info("开始执行connector start()方法");
        /*1.加载基本配置参数*/
        this.initConnectHdfsConfig(props);

        /*2.进行认证*/
        //2.1不需要认证
        if(!"2".equals(connectHdfsConfig.getHdfsLoginType())) {
            log.info("不需要进行华为云认证");
            return;
        }
        //2.2 hadoop华为云认证
        try {
            LoginUtil.login(connectHdfsConfig.getHuaweiLoginBean());
        } catch (IOException e) {
            log.error("华为云hadoop认证失败，终止本connector运行, 原因：{}", e.getMessage(), e);
            throw new RuntimeException("华为云hadoop认证失败");
        }
    }

    /**
     * 初始化connector基本配置参数
     * @param props
     */
    private void initConnectHdfsConfig(Map<String, String> props) {
        connectHdfsConfig = new ConnectHdfsConfig();
        connectHdfsConfig.setHdfsUrl(props.get(GlobalConstant.HDFS_URL));
        connectHdfsConfig.setHdfsPath(props.get(GlobalConstant.HDFS_PATH));
        connectHdfsConfig.setHdfsFileFormat(props.get(GlobalConstant.HDFS_FILE_FORMAT));
        connectHdfsConfig.setHdfsLoginType(props.get(GlobalConstant.HDFS_LOGIN_TYPE));
        if(connectHdfsConfig.getHdfsLoginType() == null) {
            connectHdfsConfig.setHdfsLoginType("0");
        }
        if(!"0".equals(connectHdfsConfig.getHdfsLoginType())) {
            HuaweiLoginBean huaweiLoginBean = new HuaweiLoginBean();
            connectHdfsConfig.setHuaweiLoginBean(huaweiLoginBean);
            String propFlushCounter = props.get(GlobalConstant.HUAWEI_AUTH_FLUSH_COUNTER);
            if(StringUtils.isEmpty(propFlushCounter)) {
                propFlushCounter = "360";
            }
            huaweiLoginBean.setFlushCounter(Integer.parseInt(propFlushCounter));
            huaweiLoginBean.setUserPrincipal(props.get(GlobalConstant.HUAWEI_USER_PRINCIPAL));
            huaweiLoginBean.setLoginUsername(props.get(GlobalConstant.HUAWEI_LOGIN_USERNAME));
            huaweiLoginBean.setUserKeytabPath(props.get(GlobalConstant.HUAWEI_USER_KEYTAB_PATH));
            huaweiLoginBean.setKrb5ConfPath(props.get(GlobalConstant.HUAWEI_KRB5_CONF_PATH));
            huaweiLoginBean.setZkServerPrincipal(props.get(GlobalConstant.HUAWEI_ZK_SERVER_PRINCIPAL));
            huaweiLoginBean.setZkServerPrincipalKey(props.get(GlobalConstant.HUAWEI_ZK_SERVER_PRINCIPAL_KEY));
            huaweiLoginBean.setZkLoginContentName(props.get(GlobalConstant.HUAWEI_ZK_LOGIN_CONTENT_NAME));
            huaweiLoginBean.setHdfsSitePath(props.get(GlobalConstant.HUAWEI_HDFS_SITE_PATH));
            huaweiLoginBean.setCoreSitePath(props.get(GlobalConstant.HUAWEI_CORE_SITE_PATH));
        }
        log.info("加载到的配置参数connectHdfsConfig:{}", JSON.toJSONString(connectHdfsConfig));
    }

    /**
     * 指定本connector对应的task
     * @return
     */
    @Override
    public Class<? extends SinkTask> taskClass() {
        log.info("开始执行taskClass()方法");
        if("parquet".equals(connectHdfsConfig.getHdfsFileFormat())) {
            return ParquetSinkTask.class;
        } else {
            return TextSinkTask.class;
        }
    }

    /**
     * 初始化task的配置文件
     * @param maxTasks
     * @return
     */
    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        log.info("开始执行taskConfigs()方法");
        List<Map<String, String>> configs = new ArrayList<Map<String, String>>();
        for (int i = 0; i < maxTasks; i++) {
            Map<String, String> config = new HashMap<String, String>();
            config.put(GlobalConstant.HDFS_URL, connectHdfsConfig.getHdfsUrl());
            config.put(GlobalConstant.HDFS_PATH, connectHdfsConfig.getHdfsPath());
            config.put(GlobalConstant.HDFS_LOGIN_TYPE, connectHdfsConfig.getHdfsLoginType());
            //华为云认证时需要该配置文件
            if(connectHdfsConfig.getHuaweiLoginBean() != null) {
                config.put(GlobalConstant.HUAWEI_HDFS_SITE_PATH, connectHdfsConfig.getHuaweiLoginBean().getHdfsSitePath());
                config.put(GlobalConstant.HUAWEI_CORE_SITE_PATH, connectHdfsConfig.getHuaweiLoginBean().getCoreSitePath());
            }
            configs.add(config);
            log.info("读取到的configs配置为:{}", JSON.toJSONString(configs));
        }
        return configs;
    }

    /**
     * 执行stop方法
     */
    @Override
    public void stop() {
        log.info("connector 执行stop()方法");
    }

    @Override
    public ConfigDef config() {
        log.info("connector执行config()方法");
        return CONFIG_DEF;
    }
}
